package com.niit.spark.sql

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

/**
 * Date:2025/5/7
 * Author：Ys
 * Description:
 */
object SparkSQL_Base {

  def main(args: Array[String]): Unit = {

    //准备环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkSQL_Base")
    val spark = SparkSession.builder().config(sparkConf).getOrCreate();
    spark.sparkContext.setLogLevel("ERROR")

    //1.DataFrame
    //1.1. 使用spark SQL 读取 json文件
    // df==> 是一张数据表
    val df: DataFrame = spark.read.json("input/user.json")
    //1.2 展示DataFrame信息
    df.show()
    //1.3 想使用SQL进行查询，需要使用dataFrame去创建一张临时表，并给临时表起名字。后续对临时表进行操作即可
    df.createOrReplaceTempView("user")
    //1.4使用临时表进行查询
    val resDF1: DataFrame = spark.sql("select * from user")
    resDF1.show()
    //1.5查看表结构
    resDF1.printSchema()
    //1.6  使用DataFrame进行指定字段查询
    val resDF2: DataFrame = spark.sql("select username from user")
    resDF2.show()
    //1.6 使用DataFrame进行指定字段查询 简写方法
    df.select("age").show()
    //1.7查看username数据，以及 age + 1
    // 需要引入DSL， spark.implicit._
    // DSL：管理结构化的数据特定领域的语言。使用DSL不需要创建临时表，直接使用DataFrame进行查询即可
    import spark.implicits._
    df.select($"username", $"age" + 1).show()
    df.select('username,'age + 1 as "newAge").show()

    //1.8 查看 age 大于 30 的数据
    df.filter($"age" > 30).show()
    //1.9 按照 age 分组， 查看数的条数
    df.groupBy("age").count().show()
    //1.10  RDD 转换 DataFrame
    val sc = spark.sparkContext
    val rdd1: RDD[Int] = sc.makeRDD((1 to 10).toList)
    val df1: DataFrame = rdd1.toDF("id")
    df1.show()
    //1.11 DataFrame 转换 RDD
    val rdd2: RDD[Row] = df1.rdd
    rdd2.collect().foreach(println)


    //2.DataSet是具有强类型的集合，需要提供对应的类型信息 ，要和样例类一起使用
    //使用样例类创建Dataset
    var list = List(Person("张三",20),Person("李四",30))
    val ds1: Dataset[Person] = list.toDS
    ds1.show()

    //使用基本数据类型创建Dataset 但是不常用
    val ds2: Dataset[Int] = List(1, 2, 3, 4, 5).toDS
    ds2.show()

    //将Dataset转换成Rdd
    val rdd3: RDD[Person] = ds1.rdd
    rdd3.collect().foreach(println)

    //Rdd(已经包含类型【样例类】) 转换成 Dataset
    val ds3: Dataset[Person] = rdd3.toDS()

    //Rdd(不包含类型【样例类】) 转换成 Dataset
    // 利用rdd中的map算子 将里面的数据转换成样例类
    val rdd4: RDD[(String,Int)] = spark.sparkContext.makeRDD(List( ("王五", 50), ("马六",60)  ))

    val rdd5: RDD[Person] = rdd4.map(v => {

      Person(v._1, v._2)

    })
    val ds4: Dataset[Person] = rdd5.toDS()

    //Dataset 转换 DataFrame
    val df3: DataFrame = ds3.toDF()

    //DataFrame 转换 Dataset
    val ds5: Dataset[Person] = df3.as[Person]



    //关闭Spark SQL
    spark.close()
  }

  //样例类
  case class Person(name:String,age:Int)

}
